package com.diorsunion.hedge.io.netty;

import com.diorsunion.hedge.common.ConnectionStatus;
import com.diorsunion.hedge.common.ServiceResult;
import com.diorsunion.hedge.io.Message;
import com.diorsunion.hedge.io.Remoter;
import com.diorsunion.hedge.io.IOServer;
import com.diorsunion.hedge.util.LogUtil;
import com.google.common.collect.Sets;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Component("nettyServer")
public class NettyServer implements IOServer {
	@Resource @Qualifier("ServerBootstrap")
	ServerBootstrap bootstrap;
	@Resource @Qualifier("BossGroup")
	NioEventLoopGroup bossGroup;
	@Resource @Qualifier("WorkerGroup")
	NioEventLoopGroup workerGroup;
	@Resource @Qualifier("NettyServerHandler")
	NettyServerHandler nettyHandler;

	volatile ChannelFuture serverChannelFuture;
	volatile Integer port;//启动后的端口号
	AtomicInteger retry = new AtomicInteger(0);//重试次数
	private static final int max_retry = 10;//最大重试次数
	private static final int retry_time =3;//重试时间间隔

	@Override
	public ServiceResult<ConnectionStatus> status() {
		if(this.port!=null){
			return ServiceResult.successResult(ConnectionStatus.OK,"Netty服务启动中,端口号:"+this.port);
		}else {
			return ServiceResult.successResult(ConnectionStatus.NoOK,"Netty服务已尚未启动");
		}
	}

	@Override
	public ServiceResult<ConnectionStatus> start(int port) {
		if(this.port!=null){
			return ServiceResult.successResult("Netty服务已经启动,端口号:"+this.port);
		}
		if(retry.intValue()>max_retry){
			retry.set(0);//重试归0，需要手工下次启动
			LogUtil.info("Starting netty overflow 10 times!");
			return ServiceResult.errorResult("Starting netty 超过10次失败,彻底失败了,起不来了");
		}
		if(retry.intValue()==0){
			LogUtil.info("netty started at " + port);
		}else {
			LogUtil.info("第"+retry.intValue()+"次重试,启动netty服务端，打开端口监听 " + port);
		}
		try {
			serverChannelFuture = bootstrap
					.bind(port)
					.addListener(f->{
						retry.addAndGet(1);
						if(f instanceof ChannelFuture){
							ChannelFuture cf = (ChannelFuture)f;
							if (cf.isSuccess()) {
								this.port = port;
								LogUtil.info("netty started success at "+port);
							} else {
								LogUtil.info("netty started fail,retry after 3 seconds..");
								cf.channel().eventLoop().schedule(() -> start(port), retry_time, TimeUnit.SECONDS);
							}
						}
					})
					.sync();
			boolean result = serverChannelFuture.isSuccess();
			return result?ServiceResult.successResult(ConnectionStatus.OK,"启动netty服务成功,端口号:"+port):ServiceResult.errorResult("启动netty服务失败,端口号:"+port);
		} catch (Throwable e) {
			LogUtil.error("netty started fail",e);
			return ServiceResult.returnResult(e.getMessage(),false,"启动netty服务失败,端口号:"+port,ConnectionStatus.NoOK);
		}
	}

	public ServiceResult stop() {
		if(port==null){
			return ServiceResult.errorResult("Netty服务尚未启动，无法停止");
		}
		ChannelFuture channelFuture = null;
		try {
			nettyHandler.closeAllChannel();
			channelFuture = serverChannelFuture.channel().close();
			serverChannelFuture.awaitUninterruptibly();
			boolean result = channelFuture.isSuccess();
			if(result){
				port = null;
			}
			LogUtil.info("stoped netty "+ (result?"success":"fail"));
			return result?ServiceResult.successResult(null):ServiceResult.errorResult(null);
		} catch (Throwable e) {
			LogUtil.error("stoped netty fail",e);
			return ServiceResult.errorResult(null);
		} finally {
			LogUtil.info("closing NioEventLoopGroup ...");
			bossGroup.shutdownGracefully();
			workerGroup.shutdownGracefully();
		}
	}

	@Override
	public Set<Remoter> allRemoters() {
		Set<Remoter> remoters = Sets.newHashSet();
		nettyHandler.channelGroup.parallelStream().forEach(x -> remoters.add(convertChannelToRemoter(x)));
		return remoters;
	}

	public static final Remoter convertChannelToRemoter(Channel channel){
		Remoter r = new Remoter();
		r.id = channel.id().asShortText();
		InetSocketAddress address = (InetSocketAddress)channel.remoteAddress();
		r.address = address.getHostName();
		r.port = address.getPort();
		r.active = channel.isActive();
		r.open = channel.isOpen();
		r.writable = channel.isWritable();
		r.registered = channel.isRegistered();
		return r;
	}

	@Override
	public Remoter getRemoter(String id) {
		Optional<Channel> channelIdOptional = nettyHandler.channelGroup.stream().filter(x->id.equals(x.id().asShortText())).findFirst();
		if(channelIdOptional!=null){
			Channel channel = channelIdOptional.get();
			if(channel!=null){
				Remoter remoter = convertChannelToRemoter(channel);
				return remoter;
			}
		}
		return null;
	}

	@Override
	public void write(Message message) {
		nettyHandler.write(message);
	}

	@Override
	public void write( String id,Message message) {
		nettyHandler.write(message,id);
	}
}
